package com.servlet;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.util.ArrayList;
import java.util.List;

import javax.servlet.ServletException;
import javax.servlet.annotation.WebServlet;
import javax.servlet.http.HttpServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.beans.ScoreInfo;
import com.beans.UserInfo;
import com.constant.Constant;
import com.dao.impl.HdfsDaoImpl;
import com.mapreduce.SortPartitionTest.SortMapper;


@WebServlet("/MR_SortServlet")
public class MR_SortServlet extends HttpServlet {
   
	protected void service(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		try {
			UserInfo user=(UserInfo)request.getSession().getAttribute("session_user");
			String userRoot=user.getUserName();
			
			// 创建一个作业
			Job job = Job.getInstance();
			
			// 设定map 相关的配置
			job.setMapperClass(TotalMapper.class);
			job.setMapOutputKeyClass(Text.class);
			job.setMapOutputValueClass(ScoreInfo.class);
	
			//指定要处理的文件是谁
			String filePath = request.getParameter("filePath");
			FileInputFormat.setInputPaths(job, new Path(Constant.HDFS_PATH +filePath));
			
			// 设定 reduce 相关的配置
			job.setReducerClass(TotalReduce.class);
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(ScoreInfo.class);
			
			
			
			//定义聚合后的结果存到什么地方
			Path savePath=new Path(Constant.HDFS_PATH+userRoot+"/part_sort_tmp");
			
			// 因为如果目标目录存在,将出错,所以可以先将目标删除
			URI uri = new URI(Constant.HDFS_PATH );
			
			FileSystem fs = FileSystem.get(uri, Constant.CONF,"root");
			fs.delete(savePath, true);

			// 保存计算结果
			FileOutputFormat.setOutputPath(job, savePath);
			
			job.waitForCompletion(true);
			
			//执行完上面的动作以后,会将每个用户的分数,进行聚合汇总,但没有排序
			//如果要排序,还要进行下面的处理
			Job job2 = Job.getInstance();
			job2.setMapperClass(SortMapper.class);
			job2.setMapOutputKeyClass(ScoreInfo.class);
			job2.setMapOutputValueClass(Text.class);
			
			FileInputFormat.setInputPaths(job2, savePath+"/part-r-00000");
			
			Path resultPath=new Path(Constant.HDFS_PATH+userRoot+"/part_sort_result");
			fs.delete(resultPath,true);
			FileOutputFormat.setOutputPath(job2,resultPath);
			
			boolean result=job2.waitForCompletion(true);
			
			//从hdfs上读出处理结果
			FSDataInputStream fsInput = fs.open(new Path(Constant.HDFS_PATH+userRoot+"/part_sort_result/part-r-00000"));
		
			List<ScoreInfo> scoreInfoList=new ArrayList<>();
			BufferedReader br =new BufferedReader(new InputStreamReader(fsInput,"UTF-8"));
			String str=null;
			while((str=br.readLine())!=null) {
				String [] data =str.split("\t");
				
				String address=data[0];
				String name=data[1];
				int score=Integer.parseInt(data[2]);
				String gender=data[3];
				String idCard=data[4];
				
				ScoreInfo info=new ScoreInfo(idCard,address,name,score,gender);
				scoreInfoList.add(info);
			}
		
			br.close();
			fsInput.close();
			fs.close();
			
			request.setAttribute("scoreInfoList", scoreInfoList);
			request.getRequestDispatcher("/mapreduce/sort-result.jsp").forward(request, response);
		}
		catch(Exception e) {
			e.printStackTrace();
		}
					
	}
	
	
	public static class TotalMapper extends Mapper<LongWritable, Text, Text, ScoreInfo> {
		ScoreInfo info = new ScoreInfo();
		Text k2 = new Text();

		protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, ScoreInfo>.Context context)
				throws IOException, InterruptedException {

			String line = value.toString();
			String[] data = line.split("\t");


			info.setIdCard(data[0]);
			info.setAddress(data[1]);
			info.setName(data[2]);
			info.setScore(Integer.parseInt(data[3]));
			info.setGender(data[4]);
			k2.set(info.getIdCard());
			context.write(k2, info);
		}
	}
	
	public static class TotalReduce extends Reducer<Text, ScoreInfo, Text, ScoreInfo> {
		int totalScore = 0;
		ScoreInfo info;

		protected void reduce(Text key, Iterable<ScoreInfo> values, Context context) throws IOException, InterruptedException {

			int i = 0;
			for (ScoreInfo o : values) {
				// 因为 其他信息也要输出去，这里假设id相同的用户，其他信息完全相同
				if (i == 0) {
					info = o;
				}
				totalScore += o.getScore();
				i++;
			}

			info.setScore(totalScore);

			context.write(key, info);
			
			totalScore=0;
		}
	}

	public static class SortMapper extends Mapper<LongWritable, Text, ScoreInfo, Text> {
		ScoreInfo info = new ScoreInfo();
		Text k2 = new Text();

		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {

			String line = value.toString();
			String[] data = line.split("\t");

			info.setIdCard(data[0]);
			info.setAddress(data[1]);
			info.setName(data[2]);
			info.setScore(Integer.parseInt(data[3].trim()));
			info.setGender(data[4]);

			k2.set(info.getIdCard());
			
			context.write(info,k2 );	
		}
	}
	
}
